package com.atguigu.flink.day08;

import com.atguigu.flink.beans.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2023/12/9
 * 该案例演示了算子状态-广播状态
 */
public class Flink08_OpeState_BroadcastState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //从指定端口读取水位信息
        SingleOutputStreamOperator<WaterSensor> wsDS = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());
        //从指定端口读取阈值信息
        SingleOutputStreamOperator<Integer> thresholdDS = env
            .socketTextStream("hadoop102", 8889)
            .map(Integer::valueOf);

        //定义广播状态描述器  用于声明广播状态中存放的数据类型结构
        MapStateDescriptor<String, Integer> mapStateDescriptor
            = new MapStateDescriptor<String, Integer>("mapStateDescriptor",String.class,Integer.class);
        //广播流并行度设置不了，就是1个
        BroadcastStream<Integer> broadcastDS = thresholdDS.broadcast(mapStateDescriptor);

        //将主流和广播流进行关联
        BroadcastConnectedStream<WaterSensor, Integer> connectDS = wsDS.connect(broadcastDS);

        //对 广联后的数据进行处理
        connectDS.process(
            new BroadcastProcessFunction<WaterSensor, Integer, String>() {
                //处理主流数据
                @Override
                public void processElement(WaterSensor value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                    //获取广播状态
                    ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                    //从广播状态中获取阈值信息
                    Integer threshold = broadcastState.get("threshold");
                    threshold = threshold==null?0 : threshold;
                    //获取当前水位值
                    Integer vc = value.vc;
                    if(vc > threshold){
                        out.collect("当前水位：" + vc +",超出了阈值" + threshold);
                    }
                }

                //处理广播流数据
                @Override
                public void processBroadcastElement(Integer value, Context ctx, Collector<String> out) throws Exception {
                    //获取广播状态
                    BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                    //将阈值信息放到广播状态中
                    broadcastState.put("threshold",value);
                }
            }
        ).printToErr();
        env.execute();
    }
}
